package streamingStudy.testStreaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object FileStream {

  def main(args: Array[String]): Unit = {

    //TODO： 使用自定义的数据源采集数据
    //1.初始化 Spark 配置信息
    val sparkConf = new SparkConf().setAppName("StreamWordCount").setMaster("local[*]")
    //2.初始化 SparkStreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))
    //3.创建自定义 receiver 的 Streaming
    val lineStream = ssc.receiverStream(new CustomerReceiver("192.168.200.201", 9999))
    //4.将每一行数据做切分，形成一个个单词
    val wordStream = lineStream.flatMap(_.split("\t"))
    //5.将单词映射成元组（word,1）
    val wordAndOneStream = wordStream.map((_, 1))
    //6.将相同的单词次数做统计
    val wordCountStream = wordAndOneStream.reduceByKey(_ + _)
    //7.打印
    wordCountStream.print()
    //8.启动 SparkStreamingContext
    ssc.start()
    ssc.awaitTermination()
  }

}
